package cn.edu.thu.laud.thrift;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;

import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.service.AbstractCassandraDaemon.CleaningThreadPool;
import org.apache.cassandra.thrift.CassandraServer;
import org.apache.cassandra.thrift.CustomTHsHaServer;
import org.apache.cassandra.thrift.CustomTNonBlockingServer;
import org.apache.cassandra.thrift.CustomTThreadPoolServer;
import org.apache.cassandra.thrift.TBinaryProtocol;
import org.apache.cassandra.thrift.TCustomNonblockingServerSocket;
import org.apache.cassandra.thrift.TCustomServerSocket;
import org.apache.hadoop.hive.common.LogUtils;
import org.apache.hadoop.hive.common.LogUtils.LogInitializationException;
import org.apache.hadoop.hive.common.ServerUtils;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.service.HiveServer.HiveServerCli;
import org.apache.hadoop.hive.service.HiveServer.HiveServerHandler;
import org.apache.thrift.protocol.TProtocolFactory;
import org.apache.thrift.server.TNonblockingServer;
import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TThreadPoolServer;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingServerTransport;
import org.apache.thrift.transport.TServerTransport;
import org.apache.thrift.transport.TTransportException;
import org.apache.thrift.transport.TTransportFactory;

import cn.edu.thu.laud.thrift.customed.CustomedLaUDProcessorFactory;
import cn.edu.thu.laud.thrift.service.LaudIfaceImpl;

/**
 * Simple class to run the thrift connection accepting code in separate
 * thread of control.
 *
 * hxd:copy from CassandraDaemon.ThriftServer
 */
public class LaUDServer extends Thread{
												      
	public static boolean includeHive=System.getProperty("include_hive","false").equals("false")?false:true;//hxd: add  hive launch switch.	
	private TServer serverEngine;
	
	private  HiveServerCli cli;//hxd: hive cli
	
	private HiveConf conf;//hxd:hive conf
	private CassandraServer cassandraServer;//hxd:我们将cassandraServer从构造函数中取出来作为一个字段了。
	/**
	 * hxd:
	 * copy from hive server.
	 */
	public void hiveinit(String[] args){
		try {
			cli = new HiveServerCli();
			cli.parse(args);
			// NOTE: It is critical to do this prior to initializing log4j, otherwise
			// any log specific settings via hiveconf will be ignored
			Properties hiveconf = cli.addHiveconfToSystemProperties();
			// NOTE: It is critical to do this here so that log4j is reinitialized
			// before any of the other core hive classes are loaded
			try {
				LogUtils.initHiveLog4j();
			} catch (LogInitializationException e) {
				LaUDDaemon.logger.warn(e.getMessage());
			}
			conf = new HiveConf(HiveServerHandler.class);
			ServerUtils.cleanUpScratchDir(conf);

			// set all properties specified on the command line
			for (Map.Entry<Object, Object> item : hiveconf.entrySet()) {
				conf.set((String) item.getKey(), (String) item.getValue());
			}

		} catch (Exception x) {
			x.printStackTrace();
		}
	}

	/**
	 * hxd:
	 * copy from public ThriftServer(InetAddress listenAddr, int listenPort),and modify it.
	 * add Hive initialize,
	 * 
	 * @param listenAddr
	 * @param listenPort
	 * @param args
	 */
	public LaUDServer(InetAddress listenAddr, int listenPort,String[] args)
	{
		//<div>hxd: add hive initialize.
		if(includeHive){
		// now we start listening for clients
		this.hiveinit(args);
		}
		//</div>
		
		//<div>hxd:load the really log properties file.
		try {
			LaUDLogUtils.initLaUDLog4j();
		} catch (cn.edu.thu.laud.thrift.LaUDLogUtils.LogInitializationException e1) {
			e1.printStackTrace();
		}
		//</div>

		// Transport
		LaUDDaemon. logger.info(String.format("Binding thrift service to %s:%s", listenAddr, listenPort));
		// Protocol factory
		TProtocolFactory tProtocolFactory = new TBinaryProtocol.Factory(true, true, DatabaseDescriptor.getThriftMaxMessageLength());

		// Transport factory
		int tFramedTransportSize = DatabaseDescriptor.getThriftFramedTransportSize();
		TTransportFactory inTransportFactory = new TFramedTransport.Factory(tFramedTransportSize);
		TTransportFactory outTransportFactory = new TFramedTransport.Factory(tFramedTransportSize);
		LaUDDaemon. logger.info("Using TFastFramedTransport with a max frame size of {} bytes.", tFramedTransportSize);

		//<div>hxd:laudImpl is a  thrift service implemention class，它等于同CassandraServer的加强版。
		final LaudIfaceImpl laudImpl= new LaudIfaceImpl(conf);
		//use LaudImpl's clientState to replace CassandraServer's clientState 
		 this.cassandraServer=new LaUDCassandraServer(laudImpl.getThreadLocalClientState());		
		 laudImpl.initCassandraIface(cassandraServer);		
		 CustomedLaUDProcessorFactory processorFactory=new CustomedLaUDProcessorFactory(null, conf, cassandraServer, laudImpl);
		//</div>

		if (DatabaseDescriptor.getRpcServerType().equalsIgnoreCase(LaUDDaemon.SYNC))
		{
			TServerTransport serverTransport;
			try
			{
				serverTransport = new TCustomServerSocket(new InetSocketAddress(listenAddr, listenPort),
						DatabaseDescriptor.getRpcKeepAlive(),
						DatabaseDescriptor.getRpcSendBufferSize(),
						DatabaseDescriptor.getRpcRecvBufferSize());
			}
			catch (TTransportException e)
			{
				throw new RuntimeException(String.format("Unable to create thrift socket to %s:%s", listenAddr, listenPort), e);
			}
			// ThreadPool Server and will be invocation per connection basis...
			TThreadPoolServer.Args serverArgs = new TThreadPoolServer.Args(serverTransport)
			.minWorkerThreads(DatabaseDescriptor.getRpcMinThreads())
			.maxWorkerThreads(DatabaseDescriptor.getRpcMaxThreads())
			.inputTransportFactory(inTransportFactory)
			.outputTransportFactory(outTransportFactory)
			.inputProtocolFactory(tProtocolFactory)
			.outputProtocolFactory(tProtocolFactory)
			// .processor(processor);
			//hxd:使用processorFactory而非processor.
			.processorFactory(processorFactory);

			ExecutorService executorService = new CleaningThreadPool(cassandraServer.clientState, serverArgs.minWorkerThreads, serverArgs.maxWorkerThreads);
			serverEngine = new CustomTThreadPoolServer(serverArgs, executorService);
			LaUDDaemon. logger.info(String.format("Using synchronous/threadpool thrift server on %s : %s", listenAddr, listenPort));
		}
		else
		{
			TNonblockingServerTransport serverTransport;
			try
			{
				serverTransport = new TCustomNonblockingServerSocket(new InetSocketAddress(listenAddr, listenPort),
						DatabaseDescriptor.getRpcKeepAlive(),
						DatabaseDescriptor.getRpcSendBufferSize(),
						DatabaseDescriptor.getRpcRecvBufferSize());
			}
			catch (TTransportException e)
			{
				throw new RuntimeException(String.format("Unable to create thrift socket to %s:%s", listenAddr, listenPort), e);
			}

			if (DatabaseDescriptor.getRpcServerType().equalsIgnoreCase( LaUDDaemon.ASYNC))
			{
				// This is single threaded hence the invocation will be all
				// in one thread.
				TNonblockingServer.Args serverArgs = new TNonblockingServer.Args(serverTransport).inputTransportFactory(inTransportFactory)
						.outputTransportFactory(outTransportFactory)
						.inputProtocolFactory(tProtocolFactory)
						.outputProtocolFactory(tProtocolFactory)
						//.processor(processor);
						//hxd:使用processorFactory而非processor.
						.processorFactory(processorFactory);

				LaUDDaemon.logger.info(String.format("Using non-blocking/asynchronous thrift server on %s : %s", listenAddr, listenPort));
				serverEngine = new CustomTNonBlockingServer(serverArgs);

			}
			else if (DatabaseDescriptor.getRpcServerType().equalsIgnoreCase(LaUDDaemon.HSHA))
			{
				// This is NIO selector service but the invocation will be Multi-Threaded with the Executor service.
				ExecutorService executorService = new JMXEnabledThreadPoolExecutor(DatabaseDescriptor.getRpcMinThreads(),
						DatabaseDescriptor.getRpcMaxThreads(),
						60L,
						TimeUnit.SECONDS,
						new SynchronousQueue<Runnable>(),
						new NamedThreadFactory("RPC-Thread"), "RPC-THREAD-POOL");
				TNonblockingServer.Args serverArgs = new TNonblockingServer.Args(serverTransport).inputTransportFactory(inTransportFactory)
						.outputTransportFactory(outTransportFactory)
						.inputProtocolFactory(tProtocolFactory)
						.outputProtocolFactory(tProtocolFactory)
						//.processor(processor);
						//hxd:使用processorFactory而非processor.
						.processorFactory(processorFactory);
				LaUDDaemon.logger.info(String.format("Using custom half-sync/half-async thrift server on %s : %s", listenAddr, listenPort));
				// Check for available processors in the system which will be equal to the IO Threads.
				serverEngine = new CustomTHsHaServer(serverArgs, executorService, Runtime.getRuntime().availableProcessors());
			}
		}
		if(includeHive){
		String msg = "Starting hive server on port " + cli.port
				+ " with " + cli.minWorkerThreads + " min worker threads and "
				+ cli.maxWorkerThreads + " max worker threads";
		HiveServerHandler.LOG.info(msg);
		if (cli.isVerbose()) {
			LaUDDaemon. logger.info(msg);
		}
		}else{
			LaUDDaemon.logger.warn("not run hive plugin..");
		}
		
	}

	public void run()
	{
		LaUDDaemon.logger.info("Listening for thrift clients...");
		serverEngine.serve();
	}

	public void stopServer()
	{
		LaUDDaemon.logger.info("Stop listening to thrift clients");
		serverEngine.stop();
	}

	public void setCassandraServer(CassandraServer cassandraServer) {
		this.cassandraServer = cassandraServer;
	}

	public LaUDCassandraServer getLaUDCassandraServer() {
		return (LaUDCassandraServer) cassandraServer;
	}

}
